package net.bgraham.collectionbatcher;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;

import junit.framework.TestCase;

import net.bgraham.collectionbatcher.ArrayListBatcher;
import net.bgraham.collectionbatcher.Batcher;
import net.bgraham.collectionbatcher.IBatchListener;
import net.bgraham.collectionbatcher.Log;

import org.junit.Before;
import org.junit.Test;

public class BatcherTestForMultipleThreads extends TestCase
{
	private static long maxTimeToWaitBetweenBatches = 900;
	private static long maxTimeToWaitBetweenUpdates = 100;
	
	static long count = 0;
	static long batches = 0;
	static long fullTimeoutCount = 0;
	static long earlyTimeoutCount = 0;
	
    @Override
	@Before
    public void setUp()
    {
    	count = 0;
    	batches = 0;
    	fullTimeoutCount = 0;
    	earlyTimeoutCount = 0;
    }

	@Test
	public static void testBatcherWith10ThreadsAnd10000UpdatesOnEachThread()
	{
		testBatcherWithMultipleThreadsUpdatingShouldHaveNoLosses(10, 10000);
	}

	@Test
	public static void testBatcherWith50ThreadsAnd5000UpdatesOnEachThread()
	{
		testBatcherWithMultipleThreadsUpdatingShouldHaveNoLosses(50, 5000);
	}
	
	@Test
	public static void testBatcherWith100ThreadsAnd1000UpdatesOnEachThread()
	{
		testBatcherWithMultipleThreadsUpdatingShouldHaveNoLosses(100, 1000);
	}
	
	private static void testBatcherWithMultipleThreadsUpdatingShouldHaveNoLosses(int numberOfThreads, int updatesPerThread)
	{
		count = 0;
		
		Log.log("testBatcherWithMultipleThreadsUpdatingShouldHaveNoLosses()");
		final List<Integer> receivedList = new ArrayList<Integer>();
		List<Integer> sentList = new ArrayList<Integer>();
		int howMany = numberOfThreads*updatesPerThread; 
		for (int i=0; i<howMany; i++)
		{
			sentList.add(new Integer(i));
		}
		
		ArrayListBatcher<Integer> batcher = new ArrayListBatcher<Integer>(
				new IBatchListener<Integer>()
				{
					public void batchCollected(Collection<Integer> batch, boolean fullTimeoutHit)
					{
						receivedList.addAll(batch);
						batches++;
						if (fullTimeoutHit)
						{
							fullTimeoutCount++;
						}
						else
						{
							earlyTimeoutCount++;
						}
					}
				},
				maxTimeToWaitBetweenBatches, maxTimeToWaitBetweenUpdates
				);

		List<ThreadToCallBatcher> threads = new ArrayList<ThreadToCallBatcher>();
		for (int x=0; x<numberOfThreads; x++)
		{
			threads.add(new ThreadToCallBatcher("" + x, batcher, (updatesPerThread * x), updatesPerThread));
		}
		for (ThreadToCallBatcher thread : threads)
		{
			thread.start();
		}

		for (ThreadToCallBatcher thread : threads)
		{
			try
			{
				thread.join();
			} catch (InterruptedException e)
			{
			}
		}

		// pause to let batcher send any remaining data and shut down nicely
		Util.doSleep(1000);
		batcher.exit();
		
		Util.printBatchStats(batches, fullTimeoutCount, earlyTimeoutCount);
		Util.printResults(receivedList.size(), sentList.size());
		assertTrue("Sizes are different", sentList.size() == receivedList.size());
		Collections.sort(receivedList);
		assertTrue("Received values are different than sent values", sentList.equals(receivedList));
		Log.log("\n");
	}

	public static class ThreadToCallBatcher extends Thread
	{
		@SuppressWarnings("unused")
		private final String name;
		private final Batcher<Integer> batcher;
		private final int startingNumber;
		private final int numberOfTimes;
		
		public ThreadToCallBatcher(String name, Batcher<Integer> batcher, int startingNumber, int numberOfTimes)
		{
			this.name = name;
			this.batcher = batcher;
			this.startingNumber = startingNumber;
			this.numberOfTimes = numberOfTimes;
		}

		@Override
		public void run()
		{
			for (int val=startingNumber; val<numberOfTimes + startingNumber; val++)
			{
				batcher.add(new Integer(val));
			}
		}
	}
	
}
